# kafka_viewer.py
import json
from confluent_kafka import Consumer

# Kafka 消费者配置
consumer_config = {
    'bootstrap.servers': '192.168.31.117:9092',
    'group.id': 'debug_consumer_group',
    'auto.offset.reset': 'earliest'
}

# 订阅的主题
topic_name = 'queen-to-slave'


def view_kafka_messages():
    consumer = Consumer(consumer_config)
    consumer.subscribe([topic_name])

    print(f"开始监听 topic: {topic_name}")
    print("=" * 50)

    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue

            if msg.error():
                print(f'Error: {msg.error()}')
                continue

            # 获取消息
            msg_value = msg.value()

            print(f"Partition: {msg.partition()}, Offset: {msg.offset()}")

            if msg_value is None:
                print("Message is empty (None)")
            else:
                try:
                    # 尝试解析为 JSON
                    decoded_msg = msg_value.decode('utf-8')
                    if decoded_msg.strip():
                        # 如果不是空字符串
                        try:
                            json_msg = json.loads(decoded_msg)
                            print("JSON Message:")
                            print(json.dumps(json_msg, indent=2, ensure_ascii=False))
                        except json.JSONDecodeError:
                            # 不是 JSON 格式，直接打印
                            print("Plain Text Message:")
                            print(decoded_msg)
                    else:
                        print("Message is empty (blank string)")
                except Exception as e:
                    print(f"Error decoding message: {e}")
                    print(f"Raw message bytes: {msg_value}")

            print("-" * 50)

    except KeyboardInterrupt:
        print("\n停止监听")
    finally:
        consumer.close()


if __name__ == "__main__":
    view_kafka_messages()
